package com.framework.components.redis;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCommands;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.exceptions.JedisException;

import com.framework.components.redis.config.RedisConfig;

public class RedisHelper {

	private transient static Log log = LogFactory.getLog(RedisHelper.class);
	private static AtomicInteger rdsSessions = new AtomicInteger();
	
	private static RedisHelper instance = null;
	private boolean isShared = false;
	private ShardedJedisPool shardedJedisPool = null;
	public JedisPool jedisPool = null;

	public static RedisHelper getInstance() {
		if (instance == null) {
			synchronized (RedisHelper.class) {
				if (instance == null) {
					instance = new RedisHelper();
				}
			}
		}
		return instance;
	}
	
	private RedisHelper() {
		List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
		String hosts = RedisConfig.getProperties("rds", true).getProperty(
				"hosts");
		int timeout = Integer.parseInt(RedisConfig.getProperties("rds", true)
				.getProperty("timeout"));
		String auth = RedisConfig.getProperties("rds", true)
				.getProperty("auth");
		JedisPoolConfig poolConfig = RedisConfig.getJedisPoolConfig(RedisConfig
				.getProperties("pool", true));
		String[] hostArray = hosts.split(",");
		if (hostArray.length > 1) 
			isShared = true;
		if (isShared) {
			log.info("Config with mutil nodes, Redis Pool Run With Sharde Mode!");
			int i = 0;
			for (String host : hostArray) {
				String[] iphostPair = host.split(":");
				if (i == 0) {
					int db = 0;
					String dbIndex = RedisConfig.getProperties("rds", true).getProperty("db");
					if (StringUtils.isNotBlank(dbIndex) && StringUtils.isNumeric(dbIndex))
						db = Integer.parseInt(dbIndex);
					log.info("Default redis db is " + db);
					jedisPool = new JedisPool(poolConfig, iphostPair[0],
							Integer.parseInt(iphostPair[1].trim()), timeout, auth, db);
				}
				JedisShardInfo shardInfo = new JedisShardInfo(iphostPair[0],
						Integer.parseInt(iphostPair[1].trim()), timeout,
						"Redis_" + (i++));
				if (!StringUtils.isBlank(auth))
					shardInfo.setPassword(auth);
				shards.add(shardInfo);
			}
			shardedJedisPool = new ShardedJedisPool(poolConfig, shards);
		} else {
			log.info("Just one node, Redis Pool Run With Normal(No-Sharde) Mode!");
			String[] iphostPair = hosts.split(":");
			int db = 0;
			String dbIndex = RedisConfig.getProperties("rds", true).getProperty("db");
			if (StringUtils.isNotBlank(dbIndex) && StringUtils.isNumeric(dbIndex))
				db = Integer.parseInt(dbIndex);
			log.info("Default redis db is " + db);
			jedisPool = new JedisPool(poolConfig, iphostPair[0],
					Integer.parseInt(iphostPair[1].trim()), timeout, auth, db);
		}
	}

	/**
	 * 是否做数据分片
	 * @return
	 */
	public boolean isShared() {
		return isShared;
	}
	
	/**
	 * 从资源池中获取资源连接
	 * 
	 * @return
	 */
	public JedisCommands getJedis() {
		int size = rdsSessions.addAndGet(1);
		if (log.isDebugEnabled())
			log.debug("打开连接，当前打开的redis连接为：" +size);
		if (size > 10) {
			log.warn("当前打开连接达到"+size+"个，请检查是否有redis连接未关闭!");
		}
		if (isShared) {
			return shardedJedisPool.getResource();
		} else {
			return jedisPool.getResource();
		}
	}
	
	/**
	 * 获取jedis资源
	 * @param key
	 * @return
	 */
	public Jedis getJedisWithOutShare() {
		int size = rdsSessions.addAndGet(1);
		if (log.isDebugEnabled())
			log.debug("打开连接，当前打开的redis连接为：" +size);
		if (size > 10) {
			log.warn("当前打开连接达到"+size+"个，请检查是否有redis连接未关闭!");
		}
		return jedisPool.getResource();
	}
	
	/**
	 * 回收jedis资源
	 * @param resource
	 */
	public void returnResource(Jedis resource) {
		int size = rdsSessions.decrementAndGet();
		if (log.isDebugEnabled())
			log.debug("关闭连接，当前打开的redis连接数：" +size);
		
		jedisPool.returnResource(resource);
	}
	
	/**
	 * 释放资源，返回池中
	 * 
	 * @param resource
	 */
	public void returnResource(JedisCommands resource) {
		if (resource != null) {
			int size = rdsSessions.decrementAndGet();
			if (log.isDebugEnabled())
				log.debug("关闭连接，当前打开的redis连接数：" +size);
			
			if (isShared) {
				shardedJedisPool.returnResource((ShardedJedis) resource);
			} else {
				jedisPool.returnResource((Jedis) resource);
			}
		}
	}
	
	/**
	 * 断开资源，避免由于网络异常获取到脏数据，返回池中
	 * 
	 * @param resource
	 */
	private void returnBrokenResource(JedisCommands resource) {
		if (resource != null) {
			int size = rdsSessions.decrementAndGet();
			if (log.isDebugEnabled())
				log.debug("关闭连接，当前打开的redis连接数：" +size);
			
			if (isShared) {
				shardedJedisPool.returnBrokenResource((ShardedJedis) resource);
			} else {
				jedisPool.returnBrokenResource((Jedis) resource);
			}
			resource = null;
		}
	}
	
	/**
	 * 避免由于网络异常获取到脏数据
	 * Handle jedisException, write log and return whether the connection is broken.
	 */
	public boolean handleJedisException(JedisException jedisException) {
	    if (jedisException instanceof JedisConnectionException) {
	        log.error("Redis connection lost.", jedisException);
	    } else if (jedisException instanceof JedisDataException) {
	        if ((jedisException.getMessage() != null) && (jedisException.getMessage().indexOf("READONLY") != -1)) {
	            log.error("Redis connection are read-only slave.", jedisException);
	        } else {
	            // dataException, isBroken=false
	            return false;
	        }
	    } else {
	        log.error("Jedis exception happen.", jedisException);
	    }
	    return true;
	}
	
	/**
	 * Return jedis connection to the pool, call different return methods depends on the conectionBroken status.
	 */
	public void closeResource(JedisCommands jedis, boolean conectionBroken) {
	    try {
	        if (conectionBroken) {
	            returnBrokenResource(jedis);
	        } else {
	            returnResource(jedis);
	        }
	    } catch (Exception e) {
	        log.error("return back jedis failed, will fore close the jedis.", e);
	        jedis = null;
	    }
	}

}
